package org.fujene.memman;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.fujene.communicate.MainService;
import org.fujene.communicate.MainServiceSettings;
import org.fujene.communicate.SearchFlags;
import org.fujene.communicate.deleter.DeleterPackage;
import org.fujene.communicate.hash.HashCompareTool;
import org.fujene.communicate.hash.Hashing;
import org.fujene.communicate.indexer.term.TermPackage;
import org.fujene.constants.Constants;
import org.fujene.logging.LogParser;
import org.fujene.memcon.MemoryController;
import org.fujene.memcon.DiskController;
import org.fujene.memoryblock.appendix.AppendixPackage;
import org.fujene.memoryblock.content.ContentPackage;
import org.fujene.memoryblock.dictionary.DictionaryPackage;
import org.fujene.memoryblock.invert.InvertIndexPackage;
import org.fujene.search.SearchResult;
import org.fujene.structure.ContentInfo;
import org.fujene.structure.LookupPointer;
import org.fujene.structure.SearchRangeTuple;
import org.msgpack.object.ArrayType;

public class MemoryManager{
	private static final int DIGLENGTH = Constants.SKIPPTR_INTERVAL;
	private static final int MEM_SKIPPTR = Constants.MEMORY_TYPE_SKIPPTR;
	private static final int MEM_SKIPCONT = Constants.MEMORY_TYPE_SKIPPTRCONT;

	// Memory / Disk controller
	private MemoryController memory;
	private DiskController disk;
	private MainService mains;

	public boolean stopFlag = false;

	private int termcounter = 0;

	private ContentInfo contentInfo;

	private Lock termLock = new ReentrantLock();
	private Lock invertLock = new ReentrantLock();

	private DeleteFilter delFilter;

	public MemoryManager(int blockSize, int blockLines, String dirPath, ContentInfo contentInfo,
		int backupInterval, int replication, int total, int cluster, int uniqueKeyIndex, int deleteKeyType,
		int deleteKeyIndex, MainService mains){
		this.mains = mains;
		this.disk = new DiskController(dirPath, blockSize);
		this.contentInfo = contentInfo;
		this.delFilter = new DeleteFilter(deleteKeyType, deleteKeyIndex);
		this.memory =
			new MemoryController(blockSize, blockLines, disk, contentInfo, backupInterval, replication,
				total, cluster, uniqueKeyIndex, delFilter);

		LogParser.fine("Memory manager started.");
	}

	public void storeTerm(TermPackage pkg){
		try{
			// Remove package's ID from deleted list
			if(delFilter.deleteKeyType() == MainServiceSettings.DELTYPE_ID){
				if(delFilter.filter(pkg.ID())) delFilter.remove(pkg.ID());
			}else if(delFilter.deleteKeyType() == MainServiceSettings.DELTYPE_MULTIFLAG){
				if(delFilter.filter(pkg.multiFlag(delFilter.deleteKeyIndex())))
					delFilter.remove(pkg.multiFlag(delFilter.deleteKeyIndex()));
			}
			TermInformation newDic = null;
			try{
				termLock.lock();
				// Store term to dictionary, skip if the term exists
				newDic = storeDictionary(pkg.term(), pkg.valueType(), pkg.cluster());
			}finally{
				termLock.unlock();
			}

			// store invert index and IDs
			try{
				invertLock.lock();
				storeInvIndex(newDic.getDictionary().getCurrent(), pkg);
			}finally{
				invertLock.unlock();
			}
		}catch(Exception e){
			e.printStackTrace();
		}
		if(termcounter++ % 100000 == 0) System.gc();
	}

	// Method to store content
	public void storeContent(String content, long ID, int index, LookupPointer[] appendix, int cluster){
		ContentInformation contTerm = null;

		contTerm = lookupContentPointer(ID, index, cluster);

		ContentPackage pkg = ContentPackage.setPackage(appendix, content).ID(ID).cluster(cluster);
		LookupPointer storePtr = memory.storeContent(pkg, index);
		contTerm.current(storePtr);

		makeSkipPtrForContent(contTerm, index);
	}

	// pick up pointer to invert indices
	public TermInformation searchTerm(String term, byte[] valueTypes, int cluster){
		int index, mask;
		TermInformation newTerm = lookupTerm(term, cluster);
		if(newTerm.getDictionary() == null) return newTerm;
		// lookup term
		for(int i = 0; i < contentInfo.contentsNum(); i++){
			index = i / 8;
			mask = 1 << (i % 8);
			if((valueTypes[index] & mask) == 0) newTerm.deleteInvIndex(i);
		}

		return newTerm;
	}

	public long[] getInvertIndex(LookupPointer ptr, SearchFlags filter){
		if(ptr == null) return null;
		InvertIndexPackage pkg = memory.lookupInvertIndex(ptr);
		ArrayList<Long> filtered = new ArrayList<Long>();
		// delete by ID or flag
		for(int i = 0; i < pkg.size(); i++){
			long nextOne = pkg.getContentPtr(i);
			if(delFilter.deleteKeyType() == MainServiceSettings.DELTYPE_ID
				&& delFilter.filter(pkg.getContentPtr(i))) continue;
			if(delFilter.deleteKeyType() == MainServiceSettings.DELTYPE_MULTIFLAG
				&& delFilter.filter(pkg.getMultiFlag(i, delFilter.deleteKeyIndex()))) continue;

			// Range filter
			for(int j = 0; j < filter.getBinaryFlags().length; j++){
				if(pkg.getBinaryFlag(i, j) ^ (filter.getBinaryFlags()[j] != 0)) continue;
			}

			for(int j = 0; j < filter.getFlagRange().length; j++){
				SearchRangeTuple condition = filter.getFlagRange()[j];
				if(condition != null && !condition.isRange(pkg.getMultiFlag(i, j))) continue;
			}
			filtered.add(Long.valueOf(nextOne));
		}
		long[] result = new long[filtered.size()];
		for(int i = 0; i < result.length; i++)
			result[i] = filtered.get(i).longValue();
		
		return result;
	}

	public SearchResult searchContent(long ID, int cluster){
		int i;
		LookupPointer priorPtr = null;
		String[] contents = new String[contentInfo.contentsNum()];
		String[] appendices = new String[contentInfo.appendixNum()];

		for(i = 0; i < contentInfo.contentsNum(); i++){
			ContentInformation ptrInfo = lookupContentPointer(ID, i, cluster);
			if(ptrInfo.getCurrent() != null){
				if(i == 0) priorPtr = ptrInfo.getCurrent();
				String content = memory.lookupContents(ptrInfo.getCurrent());
				contents[i] = content;
				if(delFilter.deleteKeyIndex() == i
					&& delFilter.deleteKeyType() == MainServiceSettings.DELTYPE_CONTENT
					&& delFilter.filter(contents[i])) return null;
			}else{
				System.out.printf("Cannot found ID %d, prev: %s next: %s%n", ID, ptrInfo.getPrevious(),
					ptrInfo.getNext());
				return null;
			}
		}

		// Delete filter
		if(priorPtr != null){
			LookupPointer[] appendixPointers = memory.lookupContentToAppendixPointer(priorPtr);
			for(i = 0; i < appendixPointers.length; i++){
				appendices[i] = memory.lookupAppendix(appendixPointers[i]).getString();
				if(delFilter.deleteKeyIndex() == i
					&& delFilter.deleteKeyType() == MainServiceSettings.DELTYPE_APPENDIX
					&& delFilter.filter(appendices[i])) return null;
			}
		}
		SearchResult result = SearchResult.makeResult(0.0, contents, appendices);

		return result;
	}

	@Deprecated
	public ArrayList<DictionaryPackage> getTermList(long hashFrom, long hashTo, int valueType){
		ArrayList<DictionaryPackage> terms = new ArrayList<DictionaryPackage>();

		// start form skip pointer
		LookupPointer firstPtr = /* memory.lookupSkipPointer(0) */null;
		LookupPointer nextPtr = firstPtr;
		DictionaryPackage queue;
		long hashCode;

		while(nextPtr != null){
			firstPtr = nextPtr;
			queue = memory.lookupDictionary(firstPtr);
			hashCode = Hashing.getTermHash(queue.getName());
			if(HashCompareTool.existsInMyNode(hashCode)) terms.add(queue);
			nextPtr = memory.lookupDictionaryNext(firstPtr);
		}

		return terms;
	}

	public int getTerms(){
		return termcounter;
	}

	public void getMemoryList(){
		memory.showMemoryElapsed();
	}

	public void initializeMemory(){
		try{
			memory.readMemory();
		}catch(Exception ignored){}
	}

	public void saveMemory(){
		memory.saveMemories();
	}

	public int getSkipPointerLen(int cluster){
		return memory.getSkipPointerLength(cluster);
	}

	private TermInformation lookupTerm(String query, int cluster){
		// Look at skip pointer first
		int skipPointerLength = memory.getSkipPointerLength(cluster);
		LookupPointer dicPtr = null;
		DictionaryPackage storedTerm = null;
		TermInformation returnTerm = null;
		int i, skipLevel = 0, skipVal = 1;

		if(skipPointerLength > 1){
			skipLevel = (int)(Math.log(skipPointerLength - 1) / Math.log(DIGLENGTH));
			skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
		}
		// Find from pointers in skip-pointer, if not, state to traversal search
		for(i = 0; i < skipPointerLength; i += skipVal){
			dicPtr = memory.lookupSkipPointer(i, cluster);
			// No pointer
			if(dicPtr == null) break;

			storedTerm = memory.lookupDictionary(dicPtr);
			int comp = storedTerm.getName().compareTo(query);
			if(comp > 0){
				// skip pointer advances -> revert pointer
				skipLevel--;
				if(skipLevel == -1 || i == 0){
					i--;
					break;
				}
				i -= skipVal;
				skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
			}else if(comp == 0) return new TermInformation(storedTerm).skipptr(i);

			// traverse
			if(i + skipVal >= skipPointerLength){
				do{
					skipLevel--;
					if(skipLevel == -1){
						skipVal = 1;
						break;
					}
					skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
				}while(i + skipVal >= skipPointerLength);
			}
		}
		if(i == -1){
			// prior to the first pointer
			if(storedTerm != null) dicPtr = storedTerm.getCurrent();
			storedTerm = null;
		}else{
			// get skip pointer again
			if(i == skipPointerLength) i--;
			if(i >= 0){
				dicPtr = memory.lookupSkipPointer(i, cluster);
				storedTerm = memory.lookupDictionary(dicPtr);
			}
		}

		// find by dictionary order
		if(storedTerm != null){
			LookupPointer prevPtr = storedTerm.getCurrent();
			DictionaryPackage nextTerm = null;
			if(storedTerm.getNext() != null) nextTerm = memory.lookupDictionary(storedTerm.getNext());
			int count = 0;
			while(nextTerm != null){
				int comp = nextTerm.getName().compareTo(query);
				if(comp > 0){
					// Not exists
					returnTerm =
						new TermInformation(null).middle(prevPtr, nextTerm.getCurrent()).skipptr(i)
							.appendSkip(count >= DIGLENGTH);
					break;
				}else if(comp == 0){
					// Found term in the dictionary
					returnTerm = new TermInformation(nextTerm).middle(prevPtr, nextTerm.getNext()).skipptr(i);
					break;
				}else{
					prevPtr = nextTerm.getCurrent();
					if(nextTerm.getNext() == null) break;
					nextTerm = memory.lookupDictionary(nextTerm.getNext());
				}
				count++;
			}

			if(returnTerm == null){
				// The final term in dictionary
				returnTerm =
					new TermInformation(null).behindLast(prevPtr).skipptr(i).appendSkip(count >= DIGLENGTH);
			}
		}else{
			// prior to first term(i = -1) or the first term(i = 0)
			returnTerm = new TermInformation(null).behindFirst(dicPtr).skipptr(i);
		}

		return returnTerm;
	}

	private void makeSkipPtrForTerm(TermInformation term){
		int index = term.getSkipIndex();

		if(index == -1){
			// Overwrite first pointer
			memory.storeSkipPointer(0, -1, term.getDictionary().getCurrent(), true, MEM_SKIPPTR, term
				.getDictionary().getCluster());
			refreshSkipPtrDict(term.getDictionary().getCluster());
		}else{
			if(term.isAppendSkipPointer()){
				// Append pointer
				memory.storeSkipPointer(index + 1, -1, term.getDictionary().getCurrent(), false, MEM_SKIPPTR,
					term.getDictionary().getCluster());
			}
		}
	}

	private void makeSkipPtrForContent(ContentInformation term, int type){
		int index = term.getSkipIndex();

		if(index == -1){
			// overwrite first pointer
			memory.storeSkipPointer(0, type, term.getCurrent(), true, MEM_SKIPCONT, term.getCurrent()
				.cluster());
			refreshSkipPtrContent(type, term.getCurrent().cluster());
		}else{
			if(term.isAppendSkipPointer()){
				// append skip pointer
				memory.storeSkipPointer(index + 1, type, term.getCurrent(), false, MEM_SKIPCONT, term
					.getCurrent().cluster());
			}
		}
	}

	private void refreshSkipPtrDict(int cluster){
		LookupPointer threshold = memory.lookupSkipPointer(1, cluster);
		LookupPointer nextPtr = memory.lookupSkipPointer(0, cluster);
		LookupPointer prevPtr = null;
		int count = 0;
		if(threshold == null){
			while(nextPtr != null){
				prevPtr = nextPtr;
				nextPtr = memory.lookupDictionaryNext(nextPtr);
				count++;
			}
		}else{
			while(nextPtr != null && !threshold.equals(nextPtr)){
				prevPtr = nextPtr;
				nextPtr = memory.lookupDictionaryNext(nextPtr);
				count++;
			}
		}

		if(count >= DIGLENGTH) memory.storeSkipPointer(1, 0, prevPtr, false, MEM_SKIPPTR, cluster);
	}

	// refresh skip pointer not to jam in the first pointer
	private void refreshSkipPtrContent(int type, int cluster){
		LookupPointer threshold = memory.lookupContentsSkipPointer(1, type, cluster);
		LookupPointer nextPtr = memory.lookupContentsSkipPointer(0, type, cluster);
		LookupPointer prevPtr = null;
		int count = 0;
		if(threshold == null){
			while(nextPtr != null){
				prevPtr = nextPtr;
				nextPtr = memory.lookupNextContentPointer(nextPtr);
				count++;
			}
		}else{
			while(nextPtr != null && !threshold.equals(nextPtr)){
				prevPtr = nextPtr;
				nextPtr = memory.lookupNextContentPointer(nextPtr);
				count++;
			}
		}

		if(count >= DIGLENGTH) memory.storeSkipPointer(1, type, prevPtr, false, MEM_SKIPCONT, cluster);
	}

	// Store an term
	private TermInformation storeDictionary(String query, int valueType, int cluster)
		throws UnsupportedEncodingException{
		TermInformation newTerm = lookupTerm(query, cluster);

		if(!newTerm.isFound()){
			// Add term if a query doesn't appear
			DictionaryPackage addPkg =
				DictionaryPackage.setPackage(query, contentInfo.contentsNum(), cluster).next(
					newTerm.getNext());
			LookupPointer newTermPtr = memory.storeDictionary(addPkg, cluster);
			newTerm.dictionary(addPkg.current(newTermPtr));

			// Reconnect pointer to the term
			if(newTerm.getPrevious() != null)
				memory.storeDictionaryPointer(newTerm.getPrevious(), newTermPtr, cluster);

			// Add and modify dictionary
			makeSkipPtrForTerm(newTerm);

			this.termcounter++;
		}else{
			memory.incrementDocFreq(newTerm.getDictionary().getCurrent(), cluster);
		}

		return newTerm;
	}

	private void storeInvIndex(LookupPointer dicPtr, TermPackage pkg){
		// pick up pointer to invert index
		dicPtr.withCluster(pkg.cluster());
		LookupPointer leadPtr = memory.lookupDictionaryInvIndex(dicPtr, pkg.valueType());
		LookupPointer allocPtr;

		// Write invert index
		if(leadPtr == null) allocPtr = memory.allocateInvertIndex(100, pkg.cluster());
		else{
			// System.out.println("[DEBUG]Lookup " + LookupPointer.valueOf(dicPtr.block(), dicPtr.point() + 9)
			// + "->" + leadPtr);
			allocPtr = leadPtr;
		}

		allocPtr.withCluster(pkg.cluster());
		LookupPointer invIndexPtr =
			memory.storeInvertIndex(allocPtr, pkg.ID(), pkg.binaryFlags(), pkg.multiFlags());

		// refresh dictionary point
		if(leadPtr == null){
			memory.storeDictionaryToInvIndex(dicPtr, invIndexPtr, pkg.valueType());
			// System.out.println("[DEBUG]Store " + LookupPointer.valueOf(dicPtr.block(), dicPtr.point() + 9)
			// + "->" + invIndexPtr);
		}
	}

	// Look up pointers specified by ID
	private ContentInformation lookupContentPointer(long ID, int subtype, int cluster){
		// look up skip pointer
		int skipPointerLength = memory.getSkipContentsLength(subtype, cluster);
		long nextID = 0;
		LookupPointer nextPtr = null;
		LookupPointer prevPtr = null;
		int i = 0;
		int skipLevel = 0;
		int skipVal = 1;
		if(skipPointerLength > 1){
			skipLevel = (int)(Math.log(skipPointerLength - 1) / Math.log(DIGLENGTH));
			skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
		}
		for(i = 0; i < skipPointerLength; i += skipVal){
			nextPtr = memory.lookupContentsSkipPointer(i, subtype, cluster);
			// No pointer
			if(nextPtr == null || nextPtr.equals(LookupPointer.EMPTY)) break;

			nextID = memory.lookupContentID(nextPtr);
			if(nextID > ID){
				// before this ID
				skipLevel--;
				if(skipLevel == -1 || i == 0){
					i--;
					break;
				}
				i -= skipVal;
				skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
			}else if(nextID == ID){
				// Just this content
				return new ContentInformation(nextPtr).ID(ID).skipptr(i);
			}

			// look at past
			if(i + skipVal >= skipPointerLength){
				do{
					skipLevel--;
					if(skipLevel == -1){
						skipVal = 1;
						break;
					}
					skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
				}while(i + skipVal >= skipPointerLength);
			}
		}

		// the first content
		if(i == -1) return new ContentInformation(null).behindFirst()
			.next(memory.lookupContentsSkipPointer(0, subtype, cluster)).appendSkip(true).skipptr(-1);
		else{
			if(i == skipPointerLength) i--;
			nextPtr = memory.lookupContentsSkipPointer(i, subtype, cluster);
		}

		// trace contents by ID
		int count = 0;
		prevPtr = nextPtr;
		while(nextPtr != null){
			long l = memory.lookupContentID(nextPtr);
			if(l > ID){
				// The content is not in the list
				return new ContentInformation(null).middle(prevPtr).next(nextPtr)
					.appendSkip(count >= DIGLENGTH).skipptr(i);
			}else if(l == ID){
				// ID matched -> nextPtr is actual existing point
				return new ContentInformation(nextPtr).ID(l).middle(prevPtr).appendSkip(false).skipptr(i);
			}else{
				prevPtr = nextPtr;
				nextPtr = memory.lookupNextContentPointer(nextPtr.withCluster(cluster));
			}
			count++;
		}
		return new ContentInformation(null).middle(prevPtr).next(nextPtr).appendSkip(count >= DIGLENGTH)
			.skipptr(i);
	}

	// Store appendix on memory
	public LookupPointer storeAppendix(String content, long ID, int index, int cluster){
		// If any content with same ID is found, it does not over write
		return memory.storeAppendix(AppendixPackage.setPackage().string(content).cluster(cluster), index);
	}

	public void deleteContent(DeleterPackage pkg){
		delFilter.add(pkg);
	}

	// backup ///////////////
	public void revertLatest(){
		memory.revert(-1);
	}

	public void revertByEntry(long ID){
		memory.revert(ID);
	}

	public long takeSnapshot(){
		return memory.snapshot();
	}

	public void backupLatest(){
		memory.snapshot();
		memory.backup();
	}

	public long latestBackupIndex(){
		return memory.latestBackupIndex();
	}

	// ///////////////////////

	public void addCluster(int myOrder, int totalNodes){
		memory.addCluster(myOrder, totalNodes);
	}

	public void synchronizeClock(){
		memory.synchronizeClock();
	}

	public void optimizeMemory(){
		memory.optimizeMemory();
	}

	public void showBackupList(){
		memory.showBackupList();
	}

	public void suspendMyNode(){
		mains.suspendMyNode(0);
	}

	public void resumeMyNode(){
		mains.resumeMyNode();
	}

	public int getClusterBlockAmount(int clusterID){
		return memory.getClusterBlockAmount(clusterID);
	}

	public void setClusterPointInfo(ArrayType info){
		memory.setClusterPointInfo(info);
	}

	public ArrayType getClusterPointInfo(int clusterID){
		return memory.getClusterPointInfo(clusterID);
	}

	public ArrayType toArrayType(int cluster, int block){
		return memory.toArrayType(cluster, block);
	}

	public void fromArrayType(int cluster, int ID, int type, int subtype, byte[] array){
		memory.fromArrayType(cluster, ID, type, subtype, array);
	}
}
